package com.property.manage.service;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class KafkaConsumer {

    private AtomicInteger ai = new AtomicInteger(1);

    ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 30, 10, TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            new ThreadFactoryBuilder().setNameFormat("KThread-%d").build(),
            (r, executor) -> {
                System.out.println("消费拒绝");
                try {
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
//                        new ThreadPoolExecutor.CallerRunsPolicy()

    );


    //kafka的监听器，topic为"zhTest"，消费者组为"zhTestGroup"
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"0"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup1(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 0);


    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"1"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 1);


    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"2"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup3(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 2);


    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"3"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup4(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 3);


    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"4"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup5(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 4);


    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"5"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup6(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 5);
    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"6"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup7(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 6);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"7"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup8(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 7);
    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"8"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup9(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 8);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"9"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup10(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 9);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"10"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup11(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 10);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"11"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup12(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 11);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"12"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup13(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 12);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"13"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup14(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 13);
    }
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "dataCenter", partitions = {"14"})}, groupId = "test-consumer-group")
    public void listenZhugeGroup15(ConsumerRecord<String, String> record, Acknowledgment ack) {
        execute(record, ack, 14);
    }

    private void execute(ConsumerRecord<String, String> record, Acknowledgment ack, Integer seq) {
        String value = record.value();
        executor.execute(() -> {
            System.out.println("分区[" + seq + "]数据: " + value);
            int i = ai.addAndGet(1);
            System.out.println("整数记录: " + i);
        });
        ack.acknowledge();
    }
    /*//配置多个消费组
    @KafkaListener(topics = "zhTest",groupId = "zhTestGroup2")
    public void listenTulingGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        ack.acknowledge();
    }*/

}
